package com.rraj.common.mq.test;

import java.util.Properties;

import javax.annotation.Resource;

import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.PropertyValueConst;
import com.aliyun.openservices.ons.api.order.OrderConsumer;
import com.rraj.common.mq.test.constant.MQTestConsumerEnum;
import com.rraj.common.mq.test.constant.MQTestTagEnum;
import com.rraj.common.mq.test.constant.MQTestTopicEnum;
import com.rraj.common.mq.test.messagelistener.MessageListenerImpl;
import com.rraj.common.mq.test.messagelistener.MessageListenerImpl1;
import com.rraj.common.mq.test.messagelistener.MessageListenerImpl2;
import com.rraj.common.mq.test.messagelistener.MessageOrderListenerImpl;
import com.rraj.common.mq.test.messagelistener.MessageOrderListenerImpl1;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Lazy;

/**
 * Created by WangSheng on 2017/6/8.
 *
 * @author WangSheng
 */
@Configuration
@Lazy
public class MQTestConsumerConfig {
    @Value("${aliyunMQ.access_key}")
    private String accessKey;
    @Value("${aliyunMQ.secret_key}")
    private String secretKey;

    @Resource
    private MessageListenerImpl messageListenerImpl;
    @Resource
    private MessageListenerImpl1 messageListenerImpl1;
    @Resource
    private MessageListenerImpl2 messageListenerImpl2;
    @Resource
    private MessageOrderListenerImpl messageOrderListenerImpl;
    @Resource
    private MessageOrderListenerImpl1 messageOrderListenerImpl1;

    /**
     * 普通消费：集群消费
     */
    @Bean
    public Consumer clusteringConsumer() {
        Consumer consumer = ONSFactory.createConsumer(
            getConsumerProperties(MQTestConsumerEnum.CONSUMER_DEFAULT_TEST_CID1.getCid()));
        String subTag = "TAG_DEFAULT_TEST_TOPIC1_SYNC || TAG_DEFAULT_TEST_TOPIC1_ASYNC || "
            + "TAG_DEFAULT_TEST_TOPIC1_ONEWAY || TAG_DEFAULT_TEST_TOPIC1_TIMING || TAG_DEFAULT_TEST_TOPIC1_DELAY";
        consumer.subscribe(MQTestTopicEnum.TOPIC_DEFAULT.getTopic(), subTag, messageListenerImpl);
        consumer.start();
        return consumer;
    }

    /**
     * 普通消费：广播消费
     */
    @Bean
    public Consumer broadcastingConsumer() {
        Properties properties = getConsumerProperties(MQTestConsumerEnum.CONSUMER_DEFAULT_TEST_CID2.getCid());
        // 广播订阅方式
        properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.BROADCASTING);
        Consumer consumer = ONSFactory.createConsumer(properties);
        String subTag = "TAG_DEFAULT_TEST_TOPIC1_SYNC || TAG_DEFAULT_TEST_TOPIC1_ASYNC || "
            + "TAG_DEFAULT_TEST_TOPIC1_ONEWAY || TAG_DEFAULT_TEST_TOPIC1_TIMING || TAG_DEFAULT_TEST_TOPIC1_DELAY";
        consumer.subscribe(MQTestTopicEnum.TOPIC_DEFAULT.getTopic(), subTag, messageListenerImpl1);
        consumer.start();
        return consumer;
    }

    /**
     * 事务消费
     */
    @Bean
    public Consumer transactionConsumer() {
        Consumer consumer = ONSFactory.createConsumer(
            getConsumerProperties(MQTestConsumerEnum.CONSUMER_TRANSACTION_TEST_CID1.getCid()));
        consumer.subscribe(MQTestTopicEnum.TOPIC_TRANSACTION.getTopic(),
            MQTestTagEnum.TAG_TRANSACTION_TEST_TOPIC1_TAG1.getValue(), messageListenerImpl2);
        consumer.start();
        return consumer;
    }

    /**
     * 全局顺序消费
     */
    @Bean
    public OrderConsumer orderAllConsumer() {
        // 在订阅消息前，必须调用 start 方法来启动 Consumer，只需调用一次即可。
        OrderConsumer consumer = ONSFactory.createOrderedConsumer(
            getConsumerProperties(MQTestConsumerEnum.CONSUMER_ALL_ORDER_TEST_CID1.getCid()));
        consumer.subscribe(MQTestTopicEnum.TOPIC_ALL_ORDER.getTopic(),
            MQTestTagEnum.TAG_ALL_ORDER_TEST_TOPIC1_TAG1.getValue(), messageOrderListenerImpl);
        consumer.start();
        return consumer;
    }

    /**
     * 分区顺序消费
     */
    @Bean
    public OrderConsumer orderShardingConsumer() {
        // 在订阅消息前，必须调用 start 方法来启动 Consumer，只需调用一次即可。
        OrderConsumer consumer = ONSFactory.createOrderedConsumer(
            getConsumerProperties(MQTestConsumerEnum.CONSUMER_SHARDING_ORDER_TEST_CID1.getCid()));
        consumer.subscribe(MQTestTopicEnum.TOPIC_SHARDING_ORDER.getTopic(),
            MQTestTagEnum.TAG_SHARDING_ORDER_TEST_TOPIC1_TAG1.getValue(), messageOrderListenerImpl1);
        consumer.start();
        return consumer;
    }

    private Properties getConsumerProperties(String consumerID) {
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.ConsumerId, consumerID);
        properties.put(PropertyKeyConst.AccessKey, accessKey);
        properties.put(PropertyKeyConst.SecretKey, secretKey);
        return properties;
    }
}
